package com.bigdata.mapreduce.topn;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

import java.io.IOException;
import java.util.Iterator;

/**
 * 输出的key：top2温度最高的日期
 * 输出的value: 日期对应的温度
 */
public class TopNReducer extends Reducer<Tkey, IntWritable, Text, IntWritable> {

    private Text rkey = new Text();
    private IntWritable rval = new IntWritable();

    @Override
    protected void reduce(Tkey key, Iterable<IntWritable> values, Reducer<Tkey, IntWritable, Text, IntWritable>.Context context) throws IOException, InterruptedException {
        // 传入的是相同年月的一组数据，这一组数据的日期和温度可能不同
        // 每次在迭代values的时候，key对象中的day、temp是会更新的
        Iterator<IntWritable> iter = values.iterator();

        // 默认是第一次遍历
        boolean flag = true;
        // 记录第一次遍历的日信息
        int day = 0;
        while (iter.hasNext()) {
            iter.next();
            if (flag) {
                /*
                第一条数据就是top 1
                调用next方法会更新key的值
                 */
                rkey.set("" + key.getYear() + "---" + key.getMonth()
                        + "---" + key.getDay() + "---" + key.getCityName());
                rval.set(key.getTemp());
                context.write(rkey, rval);
                // 更新标识，非第一次遍历不会执行if的代码块
                flag = false;
                // 记录第一条数据的日期
                day = key.getDay();
            } else {
                // 比较日，如果日相等跳过
                // 如果日期不等，写入结果，结束reduce方法
                if (day != key.getDay()) {
                    rkey.set("" + key.getYear() + "---" + key.getMonth()
                            + "---" + key.getDay() + "---" + key.getCityName());
                    rval.set(key.getTemp());
                    context.write(rkey, rval);
                    return;
                }
            }
        }


    }
}
